package com.crux.kafka2ch.middleware.clickhouse;

import com.crux.kafka2ch.dto.AccessLogDto;
import com.crux.kafka2ch.middleware.AccessLogStore;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Timestamp;

@Slf4j
@Service
public class ClickHouseAccessLogStoreImpl implements AccessLogStore {
    private final ClickHouseConfig clickHouseConfig;
    private static final String insertSql = "insert into logger_access_log(id ,create_by ,create_time ,is_deleted ,last_modified_by ,last_modified_version ,update_time ,client_ip ,client_port ,current_user ,end_date ,local_addr ,local_port ,method ,msg_id ,query_string ,request_uri ,request_url ,rid ,service_id ,session_context_id ,start_date ,status_code ,total_time) " + "values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";

    public ClickHouseAccessLogStoreImpl(ClickHouseConfig clickHouseConfig) {
        this.clickHouseConfig = clickHouseConfig;
    }


    private void psSetter(PreparedStatement ps, AccessLogDto accessLog) throws SQLException {
        ps.setString(1, accessLog.getId());
        ps.setString(2, accessLog.getCreateBy());
        ps.setTimestamp(3, new Timestamp(accessLog.getCreateTime().getTime()));
        ps.setBoolean(4, accessLog.getDeleted());
        ps.setString(5, accessLog.getLastModifiedBy());
        ps.setInt(6, accessLog.getLastModifiedVersion());
        ps.setTimestamp(7, new Timestamp(accessLog.getUpdateTime().getTime()));
        ps.setString(8, accessLog.getClientIp());
        ps.setString(9, accessLog.getClientPort());
        ps.setString(10, accessLog.getCurrentUser());
        ps.setTimestamp(11, new Timestamp(accessLog.getEndDate().getTime()));
        ps.setString(12, accessLog.getLocalAddr());
        ps.setInt(13, accessLog.getLocalPort());
        ps.setString(14, accessLog.getMethod());
        ps.setString(15, accessLog.getMsgId());
        ps.setString(16, accessLog.getQueryString());
        ps.setString(17, accessLog.getRequestUri());
        ps.setString(18, accessLog.getRequestUrl());
        ps.setString(19, accessLog.getRid());
        ps.setString(20, accessLog.getServiceId());
        ps.setString(21, accessLog.getSessionContextId());
        ps.setTimestamp(22, new Timestamp(accessLog.getStartDate().getTime()));
        ps.setInt(23, accessLog.getStatusCode());
        ps.setLong(24, accessLog.getTotalTime());
    }

    @Override
    public void save(AccessLogDto accessLog) {
        try {
            clickHouseConfig.getClickHouseJdbcTemplate().update(insertSql, ps -> psSetter(ps, accessLog));
        } catch (Exception ex) {
            log.error("插入AccessLog失败:{}", ex.getMessage(), ex);
        }
    }
}
